2026年6月21日

Pub/Sub、List 队列与 Streams

系统梳理 Redis Pub/Sub、List 队列与 Streams 的消息模型、投递语义、消费者组、PEL、ACK、故障窗口、幂等消费和工程选型边界。

第 8 章：Pub/Sub、List 队列与 Streams

1. 本章定位

Redis 提供了三类常见的消息通信方式：

Pub/Sub：面向实时广播，延迟低，但不保存消息。
List 队列：面向简单任务分发，支持阻塞消费，但可靠性机制需要自行补齐。
Streams：面向可回放日志和消费者组，提供 PEL、ACK、Claim 等可靠消费基础设施。

本章的核心不是背命令，而是理解不同消息模型的投递语义、故障窗口和工程边界。Redis Streams 能解决大量轻量异步任务问题，但它仍然不自动等同于 Kafka、RabbitMQ、Pulsar 等专业消息系统。

2. 学习目标

完成本章后，应能够：

解释 Pub/Sub 为什么是至多一次投递，以及断线、离线和慢消费者为什么会丢消息。
使用 List 构建阻塞队列，并说明 BLPOP 与 BLMOVE 的可靠性差异。
解释 Streams 的日志模型、消息 ID、消费者组和 PEL。
正确使用 XADD、XREAD、XREADGROUP、XACK、XPENDING 和 XAUTOCLAIM。
分析消费者在不同崩溃时点产生的消息丢失或重复消费。
设计“业务事务 + 消费记录”的幂等消费方案。
为 Streams 设置合理的容量、保留时间和裁剪策略。
在 Pub/Sub、List、Streams 与专业消息队列之间完成合理选型。

2.1 本章边界与跳转

本章是 Redis 消息通信能力的主章节。List 作为基础数据结构的命令和编码见[第 5 章](/blog/tech/Redis/05.Hash 与 List/)；Sorted Set 延迟调度的结构建模见[第 6 章](/blog/tech/Redis/06.Set、Sorted Set 与 GEO/)，生产级延迟任务、幂等和重试见第 19 章；Streams 的内存和性能治理可回到[第 12 章](/blog/tech/Redis/12.Redis 高性能原理/)和第 20 章排查。

3. 核心概念

3.1 三种投递语义

语义	含义	主要风险
至多一次 At-most-once	消息最多处理一次，也可能一次都没有	消息丢失
至少一次 At-least-once	通过重试尽量避免丢失，但可能处理多次	重复消费
恰好一次 Exactly-once	最终业务副作用只发生一次	需要跨系统事务或幂等状态机，不能仅靠一个 Redis 命令实现

“消息恰好读取一次”不等于“业务副作用恰好发生一次”。消费者可能已经完成数据库更新，却在发送 ACK 前崩溃。消息随后会再次投递，因此最终仍依赖业务幂等。

3.2 Pub/Sub：瞬时广播模型

发布者向 Channel 发布消息，所有在线订阅者都会收到一份副本。发布者不知道订阅者是谁，也不会等待订阅者处理完成。

Pub/Sub 的关键特征是：

不建立消息积压。
不保存离线消息。
没有 ACK。
没有重试队列。
没有消费者组。
没有消息回放。
一个订阅者处理失败，不会影响其他订阅者。

Redis 官方将 Pub/Sub 定义为至多一次投递：消息发出后不会重新发送；订阅者发生网络断开或处理错误时，消息会永久丢失。模式订阅与精确频道订阅同时匹配时，同一消息还可能被同一客户端收到多次。(Redis)

断线与自动重连

go-redis/v9 的 PubSub 会在网络错误后重连并重新订阅，但重连只能恢复后续订阅，无法补回断线期间已经发布的消息。此外，PubSub.Receive* 不允许多个 goroutine 并发调用。(GitHub)

慢消费者

Pub/Sub 不提供业务级背压。如果消费者读取速度低于发布速度：

Redis 服务器端客户端输出缓冲区可能持续增长。
超过服务器配置的限制后，订阅连接可能被断开。
消息进入 go-redis 后，还要进入客户端 Go Channel。
go-redis 的 Channel 默认缓冲 100 条；Channel 持续写满超过发送超时后，会直接丢弃消息。默认发送超时为 60 秒，可通过选项调整。(GitHub)

因此，扩大缓冲区只能推迟故障，不能解决长期生产速率高于消费速率的问题。

Cluster 中的 Pub/Sub

普通 Pub/Sub 在 Cluster 中需要传播消息。Redis 7.0 引入 SPUBLISH、SSUBSCRIBE 等 Sharded Pub/Sub 命令，把频道映射到 Slot，只在相关分片内传播，从而减少 Cluster Bus 流量。(Redis)

3.3 List：阻塞工作队列

最简单的 FIFO 队列可以使用：

生产者：RPUSH queue message
消费者：BLPOP queue timeout

BLPOP 在队列为空时阻塞连接，有消息到达后从队头删除并返回。多个消费者阻塞在同一个 List 上时，一条消息只会被其中一个消费者取走。

问题在于：弹出即删除。

BLPOP 成功
    ↓
消息已经从 Redis 删除
    ↓
消费者处理业务
    ↓
消费者在处理完成前崩溃
    ↓
消息永久丢失

使用 BLMOVE 改进可靠性

Redis 6.2+ 可以使用 BLMOVE，原子地把消息从待处理队列移动到处理中队列：

pending --BLMOVE--> processing

业务成功后，再通过 LREM 从 processing 中删除。若消费者崩溃，消息仍留在 processing，可以由巡检程序重新放回 pending。BLMOVE 替代了已经废弃的 BRPOPLPUSH 模式。(Redis)

但 List 可靠队列仍有明显不足：

没有原生消息 ID。
没有交付次数。
没有原生空闲时间。
没有消费者所有权信息。
没有 PEL。
没有方便的范围查询和回放。
LREM 需要扫描 List，相同消息内容还可能造成误删。
判断“处理中消息是否超时”需要额外的 ZSet、Hash 或业务数据库。

这正是 Streams 相比 List 的主要提升。

3.4 Streams：可保留、可回放的日志

Stream 可以看作一个按照消息 ID 排序的追加日志。每条 Entry 包含：

消息 ID + 一组 field/value

典型消息：

1740000000000-0
    event_id = evt-1001
    order_id = order-9001
    type     = OrderPaid

Stream ID 格式为：

<milliseconds>-<sequence>

第一部分通常是生成消息时的毫秒时间，第二部分用于区分同一毫秒内的多条消息。自动生成的 ID 保证在当前 Stream 内单调递增，但它不应直接取代业务 event_id。

XADD 在不涉及大量裁剪时通常是 O(1)，消息字段按照添加顺序存储。(Redis)

XREAD：日志订阅

XREAD 从指定 ID 之后读取消息，可以阻塞等待新消息。多个客户端读取相同区间时，每个客户端都能获得一份消息，这与 BLPOP 的竞争消费不同。使用 $ 表示“只接收调用之后的新消息”，但 $ 只应出现在第一次读取中，后续应使用上一次返回的消息 ID，否则可能跳过中间消息。(Redis)

Consumer Group：组内竞争、组间广播

同一个 Stream 可以创建多个 Consumer Group：

orders
 ├── inventory-group
 │    ├── inventory-1
 │    └── inventory-2
 └── notification-group
      ├── notification-1
      └── notification-2

其语义是：

不同 Group：各自独立读取整条日志，形成广播。
同一 Group 内的不同 Consumer：消息通常只分配给其中一个 Consumer，形成负载均衡。
每个 Group 有自己的读取进度和 PEL。

3.5 PEL、ACK 与 Claim

PEL 是 Pending Entries List，即“已经投递但尚未确认”的消息集合。

消费者执行：

XREADGROUP ... STREAMS orders >

Redis 会：

找到尚未投递给该 Group 的新消息。
把消息分配给当前 Consumer。
把消息 ID、Consumer、投递时间和投递次数记录到 PEL。
将消息返回给客户端。

消费者业务处理成功后执行：

XACK orders group-name message-id

XACK 只会从该 Group 的 PEL 中移除引用，不会自动从 Stream 日志中删除消息。Redis 官方文档明确指出，> 表示读取从未投递给该 Group 的新消息；其他 ID 用于读取当前 Consumer 自己尚未 ACK 的历史消息。(Redis)

XPENDING

XPENDING 可以检查：

Pending 总数。
最小和最大 Pending ID。
每个 Consumer 的 Pending 数量。
每条消息的拥有者。
消息空闲时间。
投递次数。

这些指标是消费者故障恢复和毒消息识别的基础。(Redis)

XAUTOCLAIM

当消费者崩溃后，其消息仍属于原 Consumer。其他消费者可以用 XAUTOCLAIM 接管空闲时间超过阈值的消息。

XAUTOCLAIM orders workers recovery-1 60000 0-0 COUNT 100

表示接管空闲至少 60 秒的 Pending 消息。

XAUTOCLAIM 从 Redis 6.2 开始提供，使用类似 SCAN 的游标；内部最多扫描约 COUNT × 10 条 PEL 记录，所以返回数量可能少于 COUNT。Claim 会重置消息空闲时间并增加投递次数。Redis 7.0+ 在扫描时还会清理“PEL 中有引用、但 Stream 正文已被删除”的记录。(Redis)

Redis 8.4+ 还允许通过 XREADGROUP ... CLAIM min-idle-time 在一次读取中同时尝试接管旧消息和读取新消息；经典面试和兼容性设计中仍应掌握 XPENDING + XCLAIM 或 XAUTOCLAIM。(Redis)

3.6 消费者崩溃的时间线

崩溃时点	结果	应对方式
消息尚未读取	消息仍在 Stream	其他消费者正常读取
已加入 PEL，业务尚未执行	消息未 ACK	超时后 Claim
业务执行一半失败	消息未 ACK，部分副作用可能已发生	业务事务或状态机
业务已提交，ACK 前崩溃	消息会再次投递	幂等消费
先 ACK，业务执行前崩溃	消息不会再次投递，业务永久遗漏	禁止提前 ACK
ACK 请求已执行，但客户端未收到响应	客户端不知道 ACK 是否成功	幂等处理，重新检查 PEL
XADD 已执行，但生产者未收到响应	生产者重试可能写入重复消息	固定 `event_id`；Redis 8.6+ 可用 IDMP

可靠消费最重要的顺序是：

读取消息
  → 执行业务副作用
  → 提交业务事务
  → XACK

它实现的是“至少一次”，而不是“天然恰好一次”。

3.7 消息幂等设计

方案一：数据库唯一约束

为每条业务事件生成不可变的 event_id，并在业务数据库建立唯一约束：

PRIMARY KEY (consumer_group, event_id)

在同一个数据库事务中：

BEGIN

INSERT consumed_event(consumer_group, event_id)
ON CONFLICT DO NOTHING

如果插入成功：
    执行业务更新

COMMIT

XACK

这样：

第一次消费：写入消费记录并执行业务。
业务提交后、ACK 前崩溃：消息再次投递。
第二次消费：唯一约束冲突，不再重复执行业务。
随后再次 ACK。

消费记录与业务更新必须处在同一个本地数据库事务中。否则“消费记录成功、业务更新失败”会造成业务永久漏执行。

方案二：业务状态机

例如订单状态只能按以下方向迁移：

CREATED → PAID → SHIPPED

重复收到 OrderPaid 时，如果订单已经是 PAID 或更后面的状态，可直接视为成功。

方案三：下游幂等键

调用支付、短信、库存等外部系统时，必须把 event_id 或业务流水号作为下游幂等键。仅在本地记录“调用过”无法覆盖“远程调用已成功，但本地未收到响应”的不确定窗口。

Redis SET NX 标记的局限

SET dedup:event-id 1 NX EX 86400

只能作为弱幂等或短期去重：

先写标记再执行业务：业务失败后可能永远不再执行。
先执行业务再写标记：崩溃窗口内仍可能重复。
标记有 TTL，过期后旧消息可能再次产生副作用。
Redis 标记无法与关系数据库事务原子提交。

3.8 Streams 容量与裁剪

Stream 不会因为消息被 ACK 就自动缩短。必须显式制定容量策略：

XADD orders MAXLEN ~ 100000 * field value

或：

XTRIM orders MAXLEN ~ 100000
XTRIM orders MINID ~ 1740000000000-0

MAXLEN：按消息数量保留。
MINID：按最小 ID 或时间窗口保留，Redis 6.2+。
=：精确裁剪。
~：近似裁剪，可以略微超过阈值，但通常更高效。
LIMIT：限制一次裁剪工作量，Redis 6.2+。

XTRIM 的复杂度与实际淘汰的 Entry 数量相关。由于 Stream 按宏节点组织，近似裁剪可以优先释放完整节点，减少 CPU 和内存整理开销。(Redis)

未 ACK 消息与裁剪

传统行为下，Stream 正文可能已经被裁剪，但 PEL 引用仍存在。随后读取 Pending 消息时，可能只看到消息 ID，却拿不到原始字段。

Redis 8.2+ 增加了裁剪引用控制：

KEEPREF：删除正文，保留 PEL 引用；默认行为。
DELREF：删除正文，同时删除所有 Group 中的 PEL 引用。
ACKED：只裁剪已被所有 Consumer Group 确认的消息。

ACKED 更适合可靠性优先的场景，但如果长期存在未 ACK 消息，Stream 长度可能持续高于期望阈值。(Redis)

容量应至少根据下式估算：

所需消息数 ≈ 峰值写入速率 × 最大保留时间
内存需求   ≈ 消息数 ×（消息正文 + ID + 索引 + PEL/Group 开销）

4. 命令与 Go 使用方法

4.1 redis-cli：Pub/Sub

终端一：

SUBSCRIBE order-events

终端二：

PUBLISH order-events '{"event_id":"evt-1","order_id":"order-1"}'

模式订阅：

PSUBSCRIBE order.*
PUBLISH order.created '{"order_id":"order-1"}'

Redis 7.0+ Cluster Sharded Pub/Sub：

SSUBSCRIBE order-events
SPUBLISH order-events '{"order_id":"order-1"}'

PUBLISH 返回的是服务器认为已发送到的订阅客户端数量，不代表这些客户端已经完成业务处理。在 Cluster 中，该返回值也不能直接解释为全局成功消费数量。PUBLISH 的复杂度为 O(N+M)，其中 N 为目标频道订阅者数，M 为模式订阅总数。(Redis)

4.2 redis-cli：List 队列

简单 FIFO：

RPUSH task:pending task-1 task-2
BLPOP task:pending 5

可靠队列模式：

RPUSH task:pending '{"id":"task-1","payload":"..."}'

BLMOVE task:pending task:processing LEFT RIGHT 5

# 业务成功后确认
LREM task:processing 1 '{"id":"task-1","payload":"..."}'

BLPOP 可同时监听多个 Key，其复杂度为 O(K)，K 是传入的 List 数量；队列为空且超时时返回 nil。多个客户端阻塞在同一个 Key 上时，等待时间最长的客户端优先获得下一条消息。(Redis)

4.3 redis-cli：Streams 基础操作

添加消息：

XADD orders * event_id evt-1 order_id order-1 type OrderCreated

添加并限制容量：

XADD orders MAXLEN ~ 100000 * \
  event_id evt-2 \
  order_id order-2 \
  type OrderPaid

范围查询：

XRANGE orders - +
XRANGE orders 1740000000000-0 + COUNT 100

非消费者组读取：

XREAD COUNT 10 STREAMS orders 0-0
XREAD BLOCK 5000 COUNT 10 STREAMS orders $

创建消费者组：

# 从历史第一条消息开始
XGROUP CREATE orders order-workers 0 MKSTREAM

# 只处理创建组之后的新消息
XGROUP CREATE orders new-order-workers $ MKSTREAM

消费者读取：

XREADGROUP GROUP order-workers worker-1 \
  COUNT 10 BLOCK 5000 \
  STREAMS orders >

查看 Pending：

XPENDING orders order-workers
XPENDING orders order-workers IDLE 60000 - + 100

确认：

XACK orders order-workers 1740000000000-0

接管故障消费者的消息：

XAUTOCLAIM orders order-workers recovery-1 \
  60000 0-0 COUNT 100

4.4 Go：Pub/Sub

package messaging

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"github.com/redis/go-redis/v9"
)

type Event struct {
	EventID string `json:"event_id"`
	Type    string `json:"type"`
	Payload any    `json:"payload"`
}

func PublishEvent(
	ctx context.Context,
	rdb *redis.Client,
	channel string,
	event Event,
) (int64, error) {
	payload, err := json.Marshal(event)
	if err != nil {
		return 0, fmt.Errorf("marshal event: %w", err)
	}

	cmdCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()

	n, err := rdb.Publish(cmdCtx, channel, payload).Result()
	if err != nil {
		return 0, fmt.Errorf("publish %q: %w", channel, err)
	}

	// n 只适合作为观测信息，不代表业务处理成功。
	return n, nil
}

func ConsumePubSub(
	ctx context.Context,
	rdb *redis.Client,
	channel string,
	handle func(context.Context, []byte) error,
) error {
	pubsub := rdb.Subscribe(ctx, channel)
	defer func() {
		if err := pubsub.Close(); err != nil {
			log.Printf("close pubsub: %v", err)
		}
	}()

	// 等待 Redis 确认订阅成功，避免初始化后立即发布造成竞态。
	readyCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
	_, err := pubsub.Receive(readyCtx)
	cancel()
	if err != nil {
		return fmt.Errorf("wait subscription confirmation: %w", err)
	}

	ch := pubsub.Channel(
		redis.WithChannelSize(1024),
		redis.WithChannelSendTimeout(5*time.Second),
	)

	for {
		select {
		case <-ctx.Done():
			return ctx.Err()

		case msg, ok := <-ch:
			if !ok {
				return fmt.Errorf("pubsub channel closed")
			}

			workCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
			err := handle(workCtx, []byte(msg.Payload))
			cancel()

			if err != nil {
				// Pub/Sub 没有 ACK 或重新投递能力。
				log.Printf(
					"pubsub message failed, channel=%s: %v",
					msg.Channel,
					err,
				)
			}
		}
	}
}

并发说明：

*redis.Client 可以被多个 goroutine 复用。
一个 PubSub 实例不应由多个 goroutine 并发调用 Receive*。
上述消费者是串行处理，能保持本地处理顺序，但处理较慢时容易积压。
需要并行处理时，应使用有界 worker pool，而不是每条消息直接启动一个 goroutine。
一旦并行处理，就不再保证业务完成顺序。
对不能丢失的订单、支付、库存事件，不应仅使用 Pub/Sub。

4.5 Go：List 可靠队列

package messaging

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"time"

	"github.com/redis/go-redis/v9"
)

type ListMessage struct {
	ID      string `json:"id"`
	Payload any    `json:"payload"`
}

func EnqueueList(
	ctx context.Context,
	rdb *redis.Client,
	queue string,
	msg ListMessage,
) error {
	payload, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("marshal list message: %w", err)
	}

	cmdCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()

	if err := rdb.RPush(cmdCtx, queue, payload).Err(); err != nil {
		return fmt.Errorf("rpush %q: %w", queue, err)
	}
	return nil
}

func RunReliableListWorker(
	ctx context.Context,
	rdb *redis.Client,
	pending string,
	processing string,
	handle func(context.Context, []byte) error,
) error {
	for {
		if err := ctx.Err(); err != nil {
			return err
		}

		// 客户端超时略大于 Redis 的阻塞时间。
		popCtx, cancel := context.WithTimeout(ctx, 6*time.Second)
		raw, err := rdb.BLMove(
			popCtx,
			pending,
			processing,
			"LEFT",
			"RIGHT",
			5*time.Second,
		).Result()
		cancel()

		switch {
		case errors.Is(err, redis.Nil):
			// Redis 端阻塞超时，不是异常。
			continue
		case err != nil:
			if ctx.Err() != nil {
				return ctx.Err()
			}
			return fmt.Errorf("blmove: %w", err)
		}

		workCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
		err = handle(workCtx, []byte(raw))
		cancel()

		if err != nil {
			// 不执行 LREM，消息保留在 processing 中等待巡检恢复。
			log.Printf("list message failed, raw=%q: %v", raw, err)
			continue
		}

		ackCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
		removed, err := rdb.LRem(ackCtx, processing, 1, raw).Result()
		cancel()
		if err != nil {
			return fmt.Errorf("remove processed message: %w", err)
		}
		if removed != 1 {
			return fmt.Errorf(
				"unexpected processing acknowledgement count: %d",
				removed,
			)
		}
	}
}

该实现仍需要一个巡检器，把超时滞留在 processing 中的消息重新放回 pending。由于 List 本身不记录领取时间，通常还要增加一个 ZSet：

ZSet member = message ID
score       = 领取时间

这会显著增加实现复杂度，业务发展到这一步时通常更适合改用 Streams。

4.6 Go：Streams 生产、消费、ACK 与重试

package messaging

import (
	"context"
	"errors"
	"fmt"
	"log"
	"time"

	"github.com/redis/go-redis/v9"
)

type StreamHandler func(
	ctx context.Context,
	eventID string,
	values map[string]interface{},
) error

func EnsureGroup(
	ctx context.Context,
	rdb *redis.Client,
	stream string,
	group string,
	start string,
) error {
	cmdCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()

	err := rdb.XGroupCreateMkStream(
		cmdCtx,
		stream,
		group,
		start,
	).Err()

	if err == nil || redis.HasErrorPrefix(err, "BUSYGROUP") {
		return nil
	}
	return fmt.Errorf("create stream group: %w", err)
}

func AddStreamEvent(
	ctx context.Context,
	rdb *redis.Client,
	stream string,
	eventID string,
	orderID string,
	eventType string,
) (string, error) {
	cmdCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
	defer cancel()

	id, err := rdb.XAdd(cmdCtx, &redis.XAddArgs{
		Stream: stream,
		MaxLen: 100_000,
		Approx: true,
		Values: []interface{}{
			"event_id", eventID,
			"order_id", orderID,
			"type", eventType,
		},
	}).Result()
	if err != nil {
		return "", fmt.Errorf("xadd stream event: %w", err)
	}
	return id, nil
}

func RunStreamConsumer(
	ctx context.Context,
	rdb *redis.Client,
	stream string,
	group string,
	consumer string,
	handle StreamHandler,
) error {
	for {
		readCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
		result, err := rdb.XReadGroup(readCtx, &redis.XReadGroupArgs{
			Group:    group,
			Consumer: consumer,
			Streams: []string{
				stream, ">",
			},
			Count: 16,
			Block: 2 * time.Second,
		}).Result()
		cancel()

		switch {
		case errors.Is(err, redis.Nil):
			// BLOCK 到期且没有新消息。
			continue
		case err != nil:
			if ctx.Err() != nil {
				return ctx.Err()
			}
			return fmt.Errorf("xreadgroup: %w", err)
		}

		for _, resultStream := range result {
			for _, msg := range resultStream.Messages {
				if err := processStreamMessage(
					ctx,
					rdb,
					stream,
					group,
					msg,
					handle,
				); err != nil {
					// 不 ACK，消息继续留在 PEL。
					log.Printf(
						"stream message failed, id=%s: %v",
						msg.ID,
						err,
					)
				}
			}
		}
	}
}

func processStreamMessage(
	ctx context.Context,
	rdb *redis.Client,
	stream string,
	group string,
	msg redis.XMessage,
	handle StreamHandler,
) error {
	eventID, ok := msg.Values["event_id"].(string)
	if !ok || eventID == "" {
		return fmt.Errorf("missing or invalid event_id")
	}

	workCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
	err := handle(workCtx, eventID, msg.Values)
	cancel()
	if err != nil {
		return fmt.Errorf("handle event %q: %w", eventID, err)
	}

	// 只能在业务副作用成功提交后 ACK。
	ackCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
	acked, err := rdb.XAck(
		ackCtx,
		stream,
		group,
		msg.ID,
	).Result()
	cancel()

	if err != nil {
		return fmt.Errorf("xack message %s: %w", msg.ID, err)
	}
	if acked != 1 {
		return fmt.Errorf(
			"xack message %s returned %d",
			msg.ID,
			acked,
		)
	}
	return nil
}

func RecoverPending(
	ctx context.Context,
	rdb *redis.Client,
	stream string,
	group string,
	recoveryConsumer string,
	minIdle time.Duration,
	handle StreamHandler,
) error {
	cursor := "0-0"

	for {
		claimCtx, cancel := context.WithTimeout(ctx, 3*time.Second)
		messages, nextCursor, err := rdb.XAutoClaim(
			claimCtx,
			&redis.XAutoClaimArgs{
				Stream:   stream,
				Group:    group,
				Consumer: recoveryConsumer,
				MinIdle:  minIdle,
				Start:    cursor,
				Count:    100,
			},
		).Result()
		cancel()

		if err != nil {
			if ctx.Err() != nil {
				return ctx.Err()
			}
			return fmt.Errorf("xautoclaim: %w", err)
		}

		for _, msg := range messages {
			if err := processStreamMessage(
				ctx,
				rdb,
				stream,
				group,
				msg,
				handle,
			); err != nil {
				// 失败后仍保留在 PEL。
				// Claim 已重置空闲时间，下一轮达到 minIdle 后才能再次接管。
				log.Printf(
					"recovered message failed, id=%s: %v",
					msg.ID,
					err,
				)
			}
		}

		if nextCursor == "0-0" {
			return nil
		}
		cursor = nextCursor
	}
}

关键工程要求：

consumer 名称必须唯一，例如 hostname + process UUID。
minIdle 应大于正常业务处理的最大合理时长，否则活跃消费者处理中的消息也可能被抢走。
RecoverPending 应周期运行，而不是只在进程启动时运行一次。
连续失败的消息要根据投递次数进入死信队列，避免无限重试。
一个阻塞中的 XREADGROUP 会占用一条客户端连接，连接池大小必须覆盖消费者数量，并为普通命令预留余量。
上述实现每个 Consumer 串行处理。要提高吞吐量，通常启动多个拥有不同 Consumer Name 的消费循环。
多消费者和重试会破坏业务完成顺序，不能把 Stream Group 理解为全局严格顺序执行器。

当前 go-redis 的 Stream API 提供 XAdd、XReadGroup、XAck、XPendingExt、XAutoClaim 和多种裁剪方法；其 XReadGroupArgs 还提供了 Redis 8.4+ 的 Claim 参数。(GitHub)

4.7 Go：数据库事务幂等

以下示例使用 PostgreSQL 风格的 ON CONFLICT：

package messaging

import (
	"context"
	"database/sql"
	"fmt"
)

func HandleOrderPaidOnce(
	ctx context.Context,
	db *sql.DB,
	consumerGroup string,
	eventID string,
	orderID string,
) error {
	tx, err := db.BeginTx(ctx, nil)
	if err != nil {
		return fmt.Errorf("begin transaction: %w", err)
	}
	defer func() {
		_ = tx.Rollback()
	}()

	result, err := tx.ExecContext(
		ctx,
		`
		INSERT INTO consumed_event(consumer_group, event_id)
		VALUES ($1, $2)
		ON CONFLICT (consumer_group, event_id) DO NOTHING
		`,
		consumerGroup,
		eventID,
	)
	if err != nil {
		return fmt.Errorf("insert consumed event: %w", err)
	}

	rows, err := result.RowsAffected()
	if err != nil {
		return fmt.Errorf("read affected rows: %w", err)
	}

	if rows == 0 {
		// 已处理过，作为幂等成功返回。
		return tx.Commit()
	}

	_, err = tx.ExecContext(
		ctx,
		`
		UPDATE orders
		SET paid = TRUE
		WHERE order_id = $1
		  AND paid = FALSE
		`,
		orderID,
	)
	if err != nil {
		return fmt.Errorf("update order: %w", err)
	}

	if err := tx.Commit(); err != nil {
		return fmt.Errorf("commit transaction: %w", err)
	}
	return nil
}

只有该函数成功返回后，外层 Stream 消费者才能执行 XACK。

4.8 Redis 8.6+：生产端幂等 XADD

生产者可能遇到：

XADD 已在 Redis 成功执行
    ↓
网络在响应返回前断开
    ↓
生产者不知道是否成功
    ↓
重试 XADD
    ↓
产生重复 Entry

Redis 8.6+ 的 IDMP 可以用“生产者 ID + 幂等 ID”识别近期重复提交：

XADD orders IDMP order-api evt-1001 * \
  event_id evt-1001 \
  order_id order-1

go-redis 参数：

id, err := rdb.XAdd(ctx, &redis.XAddArgs{
	Stream:       "orders",
	ProducerID:   "order-api",
	IdempotentID: "evt-1001",
	Values: []interface{}{
		"event_id", "evt-1001",
		"order_id", "order-1",
	},
}).Result()

重复使用同一组 ProducerID + IdempotentID 时，Redis 返回原 Entry ID，而不是写入新 Entry。

但该去重记录不是无限保留的。Redis 8.6+ 通过 XCFGSET 配置保留时长和每个 Producer 的最大记录数；达到容量后，旧记录可能被淘汰。因此它不能替代业务侧长期幂等。(Redis)

5. 典型业务场景

方案	适用场景	不适用场景	数据量要求	一致性要求	性能风险	可替代方案
Pub/Sub	在线配置刷新、实时通知、WebSocket 广播、缓存失效提示	订单、支付、库存、审计事件	消息短小、在线消费	可接受偶发丢失	订阅者多导致广播成本；慢消费者缓冲膨胀	Streams、Kafka、NATS JetStream
List + BLPOP	简单后台任务、低成本工作队列	需要回放、消费者组、失败统计、复杂重试	中小规模	可以接受弹出后丢失风险	阻塞连接；大 List；消费者崩溃丢消息	BLMOVE、Streams
List + BLMOVE	简单但希望减少任务丢失的工作队列	多 Group、复杂监控和重试	中小规模	至少一次，需要自建恢复	`processing` 堆积、`LREM` 扫描	Streams
Streams + XREAD	日志回放、多个独立实时读取者	组内任务负载均衡	中等数据量和保留窗口	读取者自行保存进度	每个读取者都接收一份，网络放大	Consumer Group、Kafka
Streams + Group	订单异步处理、任务调度、CDC 中转、事件驱动	超大规模长期日志、复杂路由、跨地域事件平台	受 Redis 内存与持久化能力约束	至少一次 + 业务幂等	PEL 增长、热 Stream、裁剪、主从切换	Kafka、RabbitMQ、Pulsar
专业 MQ	长期日志、超大吞吐、多分区、死信、延迟队列、复杂路由、跨地域复制	极简低延迟临时通知	可横向扩展到更大规模	取决于产品与配置	运维与基础设施复杂度更高	Redis Streams

6. 底层实现

6.1 Pub/Sub

Redis 维护频道到订阅客户端的映射，以及模式订阅集合。发布消息时，需要遍历：

目标频道的订阅客户端。
所有可能匹配的 Pattern Subscription。

因此 PUBLISH 的复杂度为：

O(N + M)

N 是频道订阅者数，M 是模式订阅总数。

Pub/Sub 不保存历史消息，所以消息正文不会形成持久积压；但每个慢客户端都会产生输出缓冲开销。它节省了日志存储成本，却把可靠性和流量控制问题交给应用。

6.2 List

Redis 的 List 主要使用 quicklist。它不是“每个元素一个传统链表节点”，而是由多个紧凑节点串联：

Redis 6.2 及以前，节点内部主要是 ziplist。
Redis 7.0+，节点内部主要是 listpack。
两端 Push/Pop 通常为 O(1)。
中间位置查找或 LREM 需要扫描元素。

这种结构兼顾：

链表在两端插入删除的效率。
listpack 的内存紧凑性。
避免每个小元素都承担完整指针和分配器开销。

Redis 官方对象编码文档同时指出，Streams 使用 radix tree 加 listpack。(Redis)

6.3 Streams

Streams 的主体采用：

radix tree / rax
    ↓
每个叶子或宏节点保存一个 listpack
    ↓
listpack 内连续存储多条 Entry

这样设计的原因是：

消息 ID 有序，radix tree 适合按 ID 查找和范围扫描。
多条相邻消息压入 listpack，可以降低对象头、指针和内存分配开销。
裁剪时可整块释放宏节点，提高批量删除效率。
范围查询不需要扫描整个 Stream。

消费者组还要维护：

Group 的最后投递位置。
Consumer 集合。
Group 级 PEL。
Consumer 级 Pending 索引。
每条 Pending 消息的最后投递时间、投递次数和当前拥有者。

因此，每增加一个 Consumer Group，都可能为同一 Stream 增加独立进度和 PEL 内存。大量 Group 或大量长期未 ACK 消息会显著增加空间成本。

6.4 主要命令复杂度

命令	条件下的复杂度	说明
`PUBLISH`	O(N+M)	频道订阅者数加模式订阅总数
`BLPOP key...`	O(K)	K 为监听的 Key 数量
`BLMOVE`	O(1)	两端移动
`LREM`	O(N)	可能扫描整个 List
`XADD`	通常 O(1)	同时大量裁剪时为 O(E)，E 为淘汰数
`XRANGE`	O(R)	R 为返回的 Entry 数
`XREAD`/`XREADGROUP`	与返回消息数相关	大 COUNT 会增加单次响应和主线程工作量
`XACK`	每个 ID 通常 O(1)	批量 ID 总成本随 ID 数增加
`XPENDING` 详细查询	与返回或扫描数相关	`IDLE` 过滤可能扫描更多 PEL
`XAUTOCLAIM`	COUNT 较小时近似 O(1)	实际最多扫描约 COUNT×10 条记录
`XTRIM`	O(E)	E 为实际淘汰的 Entry 数

6.5 版本差异

版本	相关变化
Redis 5.0+	引入 Streams、Consumer Group、PEL、`XREADGROUP`、`XACK`、`XCLAIM`
Redis 6.2+	`BLMOVE`、`XAUTOCLAIM`、`XTRIM MINID`、裁剪 `LIMIT`
Redis 7.0+	List 小节点从 ziplist 转向 listpack；Sharded Pub/Sub；`XAUTOCLAIM` 可清理正文已删除的 PEL 引用
Redis 8.2+	Streams 裁剪增加 `KEEPREF`、`DELREF`、`ACKED`；增加 `XACKDEL`、`XDELEX`
Redis 8.4+	`XREADGROUP` 增加 `CLAIM min-idle-time`
Redis 8.6+	`XADD IDMP/IDMPAUTO` 生产端幂等
Redis 8.8+	新增 `XNACK`，可主动释放或标记失败的 Pending 消息

经典面试中最重要的仍然是 Redis 5.0 的 Consumer Group 模型和 Redis 6.2 的 XAUTOCLAIM。

7. 高性能、高并发、高可用分析

7.1 高性能

批量读取

XREADGROUP COUNT 100 可以减少网络往返，但不能盲目增大：

单次返回数据量增加。
消费者暂停时间变长。
崩溃时可能重复处理更大批次。
如果整批完成后才 ACK，重放窗口更大。

更合理的方式是根据平均消息大小、处理时间和延迟目标逐步压测。

批量 ACK

可以一次确认多个 ID：

XACK orders workers id1 id2 id3

这样可以减少网络 RTT，但要确保对应业务都已经成功提交。不能因为批量方便而提前 ACK 尚未处理完成的消息。

消息大小

大消息会同时增加：

Redis 内存。
复制流量。
AOF 体积。
网络响应大小。
Go 反序列化成本。
Claim 和重复投递成本。

消息通常应保存业务标识和必要字段，而不是携带大型图片、文件或完整文档。

热 Stream

一个 Stream 是一个 Redis Key。即使有很多 Consumer，针对同一个 Stream 的写入和读取命令仍集中到一个节点或 Slot，可能形成热 Key。

扩展方式通常是业务分片：

orders:{00}
orders:{01}
...
orders:{31}

代价是失去跨分片全局顺序，并增加消费协调复杂度。

7.2 高并发

消费者扩容

同一 Group 内增加 Consumer，可以提高并行度，但吞吐量最终受以下因素限制：

Redis 节点处理能力。
Stream 热点。
下游数据库容量。
单条消息处理时长。
连接池大小。
Claim 和重试流量。

消费者数量不应无限增加。下游数据库已饱和时，继续增加消费者只会导致连接竞争、锁冲突和超时放大。

背压

Streams 保存消息，因此能把瞬时流量峰值转化为 Lag；但这不代表没有背压问题。Lag 持续增长时，需要：

降低生产速率。
扩容消费者。
优化下游。
暂停非核心消息。
设置容量保护。
防止 Stream 把 Redis 内存耗尽。

重试风暴

数据库故障时，如果所有 Pending 消息立即被多个恢复消费者反复 Claim，会产生：

Redis Claim 压力。
下游重复请求。
日志爆量。
网络和连接池耗尽。

应采用：

合理的 minIdle。
指数退避。
最大投递次数。
死信队列。
全局并发限制。
熔断和降级。

7.3 高可用

Redis 默认使用异步复制。主节点接受 XADD、XACK 或 List 写入后，可能在尚未复制到副本时发生故障转移。因此：

已返回成功的消息可能丢失。
已完成的 ACK 可能丢失，导致消息再次投递。
尚未复制的 PEL 状态可能回退。
生产者在超时后重试可能产生重复消息。

WAIT 可以降低故障转移丢写概率，但不会把 Redis 变成强一致系统，已确认写入在特定故障场景下仍可能丢失。(Redis)

高可用分析应区分：

能力	解决的问题
Streams PEL	消费者进程崩溃恢复
RDB/AOF	Redis 进程或机器重启后的数据恢复
主从/Sentinel/Cluster	节点故障后的服务可用性
业务幂等	重复消费和不确定结果
数据库 Outbox	数据库事务与消息生产的一致性

这些机制不能互相替代。

8. 常见错误与生产事故

8.1 用 Pub/Sub 传递支付成功事件

现象：服务滚动发布期间，部分支付成功订单没有更新权益。
根因：订阅者短暂离线，Pub/Sub 不保存离线消息。
排查方法：对比支付流水、消息发布日志和订阅服务重启时间。
修复方案：改用 Streams 或专业 MQ，并补做数据对账。
如何预防：只有允许丢失的实时提示才能使用 Pub/Sub。

8.2 go-redis 订阅 Channel 被写满

现象：Redis 端发布成功，应用却偶发缺少消息；日志出现 Channel Full 或消息被丢弃。
根因：消息处理在订阅读取循环中同步执行，处理速度低于发布速度。
排查方法：检查 go-redis 日志、处理耗时、Channel 深度和消息发布速率。
修复方案：使用有界 worker pool；缩短处理路径；必要时改用 Streams。
如何预防：监控接收速率与处理速率，不把增大缓冲区当作最终方案。

8.3 使用 BLPOP 后消费者崩溃

现象：队列长度减少，但对应任务从未完成。
根因：BLPOP 返回消息时已经从 List 删除。
排查方法：对比生产任务数、List 消费数和最终业务记录。
修复方案：改为 BLMOVE pending processing，成功后 LREM。
如何预防：重要任务必须使用确认机制或 Streams。

8.4 先 XACK 再更新数据库

现象：PEL 很干净，但部分订单没有执行相应业务。
根因：消费者提前 ACK，随后数据库更新失败或进程崩溃。
排查方法：检查消息 ID、ACK 日志和数据库事务日志的时间顺序。
修复方案：数据库提交后才能 ACK；通过对账重新补偿。
如何预防：代码审查明确“副作用成功提交 → ACK”的固定顺序。

8.5 消费成功但没有 ACK

现象：PEL 持续增长；重启后大量消息被重复执行。
根因：业务处理成功，但 ACK 错误被忽略，或部分代码分支提前返回。
排查方法：检查 XPENDING、投递次数、Consumer 分布和 ACK 错误日志。
修复方案：所有成功路径统一进入 ACK；ACK 错误必须上报。
如何预防：监控 PEL 总量、最老 Pending 空闲时间及 ACK 失败率。

8.6 minIdle 配置过小

现象：一个长任务还在运行，另一个消费者已经 Claim 并再次执行。
根因：minIdle 小于正常最大处理时间。
排查方法：对比业务耗时、Claim 时间和消息投递次数。
修复方案：提高 minIdle；拆分长任务；增加处理心跳或租约机制。
如何预防：按 P99/P999 处理时长设置阈值，并留出网络和 GC 抖动空间。

8.7 裁剪了未确认消息

现象：PEL 中存在消息 ID，但重新读取时消息字段为空或正文不存在。
根因：MAXLEN 裁剪删除了 Stream 正文，却保留了 PEL 引用。
排查方法：对比 XPENDING、XRANGE、XINFO STREAM FULL。
修复方案：扩大保留窗口；Redis 8.2+ 评估 ACKED；修复长期不 ACK 的消费者。
如何预防：保留时间必须大于最大故障恢复窗口，持续监控最老 Pending。

8.8 毒消息无限重试

现象：同一消息投递次数持续增加，日志和数据库请求暴涨。
根因：消息格式错误或业务条件永远无法满足，却没有最大重试次数。
排查方法：通过 XPENDING 查看高投递次数消息并检查载荷。
修复方案：超过阈值后写入死信 Stream，再 ACK 原消息。
如何预防：区分可重试错误、不可重试错误和系统错误；建立 DLQ 处理流程。

9. 方案选型与权衡

维度	Pub/Sub	List + BLPOP	List + BLMOVE	Streams Group	专业 MQ
消息保存	否	消费前保存	pending/processing 保存	是，可配置保留	是
离线消费	不支持	支持	支持	支持	支持
广播	原生	需复制多份队列	需复制多份队列	多 Group	原生 Topic/Subscription
组内竞争	不支持	支持	支持	原生 Consumer Group	原生
ACK	无	弹出即删除	手工 `LREM`	原生 `XACK`	原生
崩溃恢复	无	无	自建巡检	PEL + Claim	原生
回放	无	困难	困难	支持	通常支持
消息元数据	很少	自定义	自定义	ID、Owner、Idle、Delivery Count	通常丰富
顺序	单连接接收顺序	单队列弹出顺序	单队列移动顺序	投递有序，完成顺序不保证	依赖产品和分区
容量	无历史积压	Redis 内存	Redis 内存	Redis 内存及持久化	通常支持更大磁盘容量
运维复杂度	低	低	中	中	中到高
推荐用途	瞬时通知	非关键简单任务	中小型可靠任务	中等规模事件与任务	核心消息平台

选型原则：

允许丢失且只关心在线状态：Pub/Sub。
非常简单的一次性任务分发：List。
需要 ACK、重试、回放、消费者组：Streams。
需要超大规模长期日志、多分区、复杂路由或成熟治理：专业 MQ。

10. 高频面试题

10.1 Redis Pub/Sub 会不会丢消息？

问题： Redis Pub/Sub 是否可靠？

推荐回答（现场版）： Pub/Sub 是至多一次投递，不保存消息，也没有 ACK。订阅者离线、断线或来不及处理时，消息都可能永久丢失，因此只适合允许丢失的实时通知。

详细展开： 客户端重连和重新订阅只能恢复后续消息，不能恢复断线窗口。慢消费者还可能因为服务器输出缓冲或 go-redis Channel 满而丢消息。(Redis)

面试官追问：

自动重连后能否补消息？
模式订阅是否可能重复？
Cluster 中如何降低广播成本？

常见错误回答： “只要 Redis 不宕机就不会丢”“go-redis 会自动重连，所以消息可靠”。

评分点：

初级：知道没有持久化。
中级：能说明断线、慢消费者和无 ACK。
高级：能进一步说明 Sharded Pub/Sub、客户端缓冲和使用边界。

10.2 BLPOP 为什么可能丢任务？

问题： 使用 BLPOP 做消息队列有什么问题？

推荐回答（现场版）： BLPOP 返回消息时已经把它从 List 删除。如果消费者在业务完成前崩溃，Redis 中已没有该消息，因此会丢任务。

详细展开： 可以使用 Redis 6.2+ 的 BLMOVE 原子地将消息从 pending 移到 processing，成功后再 LREM，但超时恢复、投递次数和死信仍需自行实现。

面试官追问：

BRPOPLPUSH 和 BLMOVE 的关系是什么？
如何找出超时的 processing 消息？
为什么最终可能还是要换 Streams？

常见错误回答： “List 中的命令是原子的，所以不会丢消息。”

评分点：

初级：知道弹出即删除。
中级：能给出 processing List 方案。
高级：能指出 LREM、重复值、超时索引和恢复机制的复杂度。

10.3 XREAD 与 XREADGROUP 有什么区别？

推荐回答（现场版）： XREAD 类似从日志指定位置读取，多个客户端可以各自收到同一消息；XREADGROUP 使用 Consumer Group，把新消息分配给组内某个 Consumer，并通过 PEL 和 ACK 记录处理状态。

面试官追问：

不同 Group 能否读取同一条消息？
> 表示什么？
用 $ 有什么风险？

常见错误回答： “XREADGROUP 只是 XREAD 的阻塞版本。”

评分点：

初级：知道一个有 Group，一个没有。
中级：能说明组间广播、组内竞争。
高级：能说明读取进度、PEL 和 $ 只能用于第一次读取。

10.4 什么是 PEL？

推荐回答（现场版）： PEL 是某个 Consumer Group 中已投递但尚未 ACK 的消息索引，记录消息 ID、当前 Consumer、空闲时间和投递次数，用于故障恢复。

详细展开： PEL 不是消息正文的完整副本。正文仍在 Stream 中，因此过早裁剪 Stream 可能留下只有 PEL 引用、没有载荷的消息。XACK 会从 PEL 删除引用，但不删除 Stream Entry。(Redis)

面试官追问：

PEL 是 Group 级还是 Consumer 级？
ACK 是否会删除消息正文？
PEL 持续增长说明什么？

常见错误回答： “PEL 就是一个失败消息队列。”

评分点：

初级：知道是未 ACK 消息。
中级：能解释 Owner、Idle、Delivery Count。
高级：能联系裁剪、内存、监控与 Claim。

10.5 Streams 如何实现至少一次投递？

推荐回答（现场版）： 消费者读取后消息进入 PEL，业务成功后才 ACK；消费者崩溃时，消息仍在 PEL，可以由其他消费者 Claim 并再次处理。因此降低了消息丢失风险，但会产生重复消费。

面试官追问：

哪个崩溃窗口会产生重复？
先 ACK 再处理会怎样？
NOACK 有什么含义？

常见错误回答： “只要使用 XACK 就是恰好一次。”

评分点：

初级：知道 ACK。
中级：能描述崩溃恢复。
高级：能准确说明至少一次与业务幂等的关系。

10.6 为什么业务成功后仍可能重复消费？

推荐回答（现场版）： 因为业务数据库提交和 Redis XACK 是两个独立操作。数据库提交后、ACK 前发生进程崩溃，消息会再次投递。

详细展开： 这不是 Redis 特有问题，而是跨系统提交的一般一致性问题。典型解决方式是消费记录与业务更新在同一个数据库事务中提交，并通过唯一 event_id 去重。

面试官追问：

Redis SET NX 能否解决？
调用外部 HTTP 服务如何幂等？
消费记录何时清理？

常见错误回答： “把 ACK 放到 defer 中即可。”

评分点：

初级：知道可能重复。
中级：提出唯一约束。
高级：能分析本地事务、外部副作用、状态机和记录生命周期。

10.7 XCLAIM 与 XAUTOCLAIM 有什么区别？

推荐回答（现场版）： XCLAIM 通常需要先知道具体消息 ID；XAUTOCLAIM 从 Redis 6.2 开始提供，可以按最小空闲时间扫描 PEL 并批量接管，更适合消费者崩溃恢复。

详细展开： XAUTOCLAIM 返回下一次扫描游标，返回条数可能少于 COUNT；Claim 会改变 Owner、重置 Idle，并通常增加投递次数。(Redis)

面试官追问：

minIdle 太小会怎样？
JUSTID 有什么影响？
Redis 5.0 如何实现自动恢复？

常见错误回答： “XAUTOCLAIM 会自动持续后台重试。”

评分点：

初级：知道用来转移消息。
中级：能说明空闲时间和游标。
高级：能分析抢占活跃任务、投递计数和毒消息。

10.8 Streams 如何设计死信队列？

推荐回答（现场版）： 通过 XPENDING 或 Claim 结果检查投递次数，超过阈值后将原消息和失败原因写入独立 DLQ Stream，随后 ACK 原消息，避免无限重试。

详细展开： 写 DLQ 与 ACK 原消息之间仍有不确定窗口。严格要求下可使用 Lua 或事务把两个操作组合，但 Cluster 中相关 Key 必须位于同一 Slot。

面试官追问：

哪些错误不应重试？
DLQ 消息如何重新投递？
DLQ 写成功但 ACK 失败会怎样？

常见错误回答： “失败三次直接删除即可。”

评分点：

初级：知道最大重试次数。
中级：能设计 DLQ。
高级：能分析 DLQ 与 ACK 的原子性和运维闭环。

10.9 Streams 如何控制内存？

推荐回答（现场版）： 使用 XADD MAXLEN ~ N 或 XTRIM MAXLEN/MINID 控制保留量，同时确保保留窗口大于最大消费延迟和故障恢复时间。

详细展开： 近似裁剪通常比精确裁剪成本低。Redis 8.2+ 可以使用 ACKED 避免裁剪尚未被所有 Group 确认的消息，但这可能导致长度超过目标值。(Redis)

面试官追问：

ACK 后消息会自动删除吗？
MAXLEN 和 MINID 如何选择？
为什么 PEL 中可能只有 ID 没有正文？

常见错误回答： “消费者 ACK 后 Redis 会释放消息内存。”

评分点：

初级：知道需要裁剪。
中级：能解释近似裁剪。
高级：能联系 PEL、多个 Group 和 Redis 8.2 引用模式。

10.10 Streams 能保证消息顺序吗？

推荐回答（现场版）： Stream Entry 按 ID 有序，单消费者串行读取时可以保持读取顺序；但多个 Consumer 并行处理、失败重试和 Claim 后，业务完成顺序不能保证。

面试官追问：

如何保证同一订单的事件顺序？
分片 Stream 会带来什么代价？
能否只让一个 Consumer 工作？

常见错误回答： “Streams 的 ID 有序，所以业务一定严格有序。”

评分点：

初级：知道消息 ID 有序。
中级：区分读取顺序和完成顺序。
高级：提出按业务 Key 分区、单分区串行和状态版本校验。

10.11 go-redis 消费者有哪些并发注意事项？

推荐回答（现场版）： 普通 redis.Client 可以并发复用，但一个 PubSub 的 Receive* 不能被多个 goroutine 并发调用。每个阻塞消费会占用连接，Stream Consumer 应使用唯一 Consumer Name，并通过有界并发控制处理速度。

面试官追问：

连接池大小如何估算？
一个 Consumer 读取后分发给多个 worker 是否可行？
为什么不能每条消息直接启动 goroutine？

常见错误回答： “go-redis 有连接池，因此阻塞命令不占连接。”

评分点：

初级：知道客户端有连接池。
中级：知道阻塞命令占连接。
高级：能分析 Worker 数、PoolSize、背压、Owner 和关闭流程。

10.12 Redis Streams 可以替代 Kafka 或 RabbitMQ 吗？

推荐回答（现场版）： 不能一概而论。Streams 适合已有 Redis、数据量中等、追求低延迟且需要基本 ACK、重试和消费者组的场景；超大规模长期日志、多分区扩展、复杂路由、跨地域复制和成熟消息治理通常更适合专业 MQ。

面试官追问：

Streams 相比 Kafka 缺少什么？
RabbitMQ 的路由模型有什么优势？
什么规模下应迁移？

常见错误回答： “Streams 有消费者组，所以完全等价于 Kafka。”

评分点：

初级：知道产品定位不同。
中级：能比较内存、磁盘、分区、路由。
高级：能结合吞吐、保留周期、RPO/RTO、运维成本和迁移复杂度选型。

10.13 生产者重试 XADD 会不会重复？

推荐回答（现场版）： 会。Redis 已执行 XADD、但响应在网络中丢失时，生产者无法判断结果，直接重试会新增另一条 Entry。应复用稳定的业务 event_id，消费者继续做幂等；Redis 8.6+ 还可以使用 XADD IDMP 做有限窗口内的生产端去重。

面试官追问：

IDMP 能否替代消费者幂等？
幂等记录会永久保存吗？
数据库更新和 XADD 如何保持一致？

常见错误回答： “XADD 是原子命令，所以重试不会重复。”

评分点：

初级：知道超时结果不确定。
中级：提出业务 event ID。
高级：能进一步说明 Outbox、IDMP 保留窗口和端到端幂等。

11. 一分钟面试回答

Redis 中 Pub/Sub、List 和 Streams 对应三种不同消息模型。Pub/Sub 是瞬时广播，至多一次投递，没有持久化、ACK 和离线消息，适合允许丢失的实时通知。List 可以通过 RPUSH + BLPOP 实现阻塞工作队列，但弹出后消费者崩溃会丢消息；Redis 6.2+ 可以用 BLMOVE 把消息移到 processing List，成功后再删除，不过超时恢复和重试元数据仍需自建。Streams 是可保留的有序日志，Consumer Group 负责组内竞争消费，消息读取后进入 PEL，业务成功后通过 XACK 确认，消费者崩溃后可以用 XAUTOCLAIM 接管。这个模型提供的是至少一次投递，因此数据库提交后、ACK 前崩溃会导致重复消费，必须通过稳定的 event ID、数据库唯一约束和业务状态机实现幂等。容量上需要用 MAXLEN 或 MINID 裁剪，并监控 Lag、PEL、最老 Pending 和投递次数。中等规模异步任务可选 Streams，超大规模长期日志、复杂路由和成熟消息治理应选择专业 MQ。

12. 本章总结

Pub/Sub 的优势是简单、低延迟和广播，代价是不保留消息，也没有可靠消费机制。
List 的阻塞操作适合简单工作队列，但 BLPOP 存在弹出后崩溃丢失窗口。
BLMOVE 可以构建 pending/processing 模式，但恢复、超时、统计和死信仍需自建。
Streams 将消息正文、读取位置、Consumer Group 和 PEL 组合成较完整的可靠消费基础。
XACK 必须发生在业务事务提交之后。
至少一次投递必然要求消费者幂等。
XAUTOCLAIM 解决的是消费者所有权转移，不会自动解决毒消息和业务重复。
ACK 不会自动删除 Stream 正文，必须设计裁剪与保留策略。
Redis 主从复制默认异步，Streams 并不天然提供强一致消息保证。
Redis Streams 是有明确边界的轻量消息方案，而不是所有专业 MQ 的无条件替代品。

13. 自测清单

为什么 Pub/Sub 客户端自动重连后仍可能缺少消息？
go-redis 的 Pub/Sub Channel 写满后会发生什么？
BLPOP 和 BLMOVE 的故障窗口分别是什么？
为什么 List reliable queue 通常还需要 ZSet 或额外元数据？
XREAD、XREADGROUP 和 XRANGE 的用途分别是什么？
PEL 中保存哪些信息？XACK 又删除了什么？
消费者在数据库提交后、ACK 前崩溃会发生什么？
如何使用数据库唯一约束实现幂等消费？
XAUTOCLAIM 的 minIdle 设置过小会产生什么问题？
为什么 Streams 的 MAXLEN 不能只根据正常消费速率设置？

14. 官方资料

Redis Pub/Sub 模型、投递语义和 Sharded Pub/Sub。(Redis)
PUBLISH 命令及复杂度。(Redis)
BLPOP 阻塞行为与消费者调度。(Redis)
BLMOVE 与可靠 List 队列模式。(Redis)
XADD、消息 ID、裁剪和 Redis 8.6 IDMP。(Redis)
XREAD 阻塞读取、Fan-out 与 $ 的使用。(Redis)
XREADGROUP、Consumer Group、PEL、ACK 和 Redis 8.4 Claim。(Redis)
XPENDING 的摘要和详细信息。(Redis)
XAUTOCLAIM 的扫描、接管与投递次数规则。(Redis)
XTRIM、近似裁剪和 Redis 8.2 引用处理模式。(Redis)
Redis List 与 Stream 的对象编码。(Redis)
Redis 异步复制与故障转移丢写边界。(Redis)
go-redis List、Streams 与 Pub/Sub 官方实现。(GitHub)